package replay

import (
	"bytes"
	"context"
	"fmt"

	bytedtracer "code.byted.org/bytedtrace/interface-go"
	"code.byted.org/bytertc/whiteboard_server/biz/api"
	v1 "code.byted.org/bytertc/whiteboard_server/biz/api/pb/v1"
	"code.byted.org/bytertc/whiteboard_server/biz/config"
	"code.byted.org/bytertc/whiteboard_server/biz/dal"
	"code.byted.org/bytertc/whiteboard_server/biz/dal/ddl"
	"code.byted.org/bytertc/whiteboard_server/biz/dal/spec"
	"code.byted.org/bytertc/whiteboard_server/biz/pkg/object_storage"
	"code.byted.org/bytertc/whiteboard_server/biz/pkg/object_storage/s3"
	"code.byted.org/bytertc/whiteboard_server/biz/pkg/trace"
	"code.byted.org/gopkg/env"
	"code.byted.org/gopkg/logs/v2/log"
	"github.com/pkg/errors"
)

const packetsPerFile = 512

type Compactor interface {
	HandlePacket(ctx context.Context, unique spec.BoardUnique, packet *v1.Packet)
	GenerateFile(ctx context.Context, unique spec.RoomUnique, id int64) error
	InitConsoleConfig(ctx context.Context, unique spec.RoomUnique) bool
}

func New(ctx context.Context, repo dal.Repo, config config.DynamicConfig) Compactor {
	return &compactor{
		repo:   repo,
		config: config,
	}
}

type compactor struct {
	repo      dal.Repo
	config    config.DynamicConfig
	accountID string
	bucket    string
	region    string
}

func (c *compactor) InitConsoleConfig(ctx context.Context, unique spec.RoomUnique) bool {
	uniq := &api.SwitchStatusRequest{
		AppID:    unique.GetAppId(),
		Category: dal.Category_Switch_Record,
	}
	status, err := c.repo.GetSwitchFromCache(ctx, uniq)
	if err != nil {
		log.V1.CtxError(ctx, "GetSwitchFromCache err%+v,appId", err, uniq.GetAppId())
		return false
	}
	if status == dal.Switch_OFF {
		return false
	}
	uniqItem := &api.GetConsoleInfoRequest{
		AppID:    unique.GetAppId(),
		Category: dal.Category_Signal_Record,
	}
	configs, err := c.repo.GetConfig(ctx, uniqItem, false)
	if err != nil {
		log.V1.CtxError(ctx, "c.repo.GetConfig err:%+v,appId:%s", err, uniqItem.AppID)
		return false
	}
	if len(configs) == 0 {
		return false
	}
	//todo support inner tos in oc env
	c.accountID = configs[0].AccountId
	c.bucket = configs[0].Space
	c.region = configs[0].Region
	return true
}

func (c *compactor) HandlePacket(ctx context.Context, unique spec.BoardUnique, packet *v1.Packet) {
	if len(packet.Actions) == 0 {
		log.V1.CtxInfo(ctx, "no action in packet %d", packet.PacketId)
		return
	}

	if !c.InitConsoleConfig(ctx, unique) {
		return
	}

	id, total, err := c.repo.SavePacket(ctx, unique, packet)
	if err != nil {
		log.V1.CtxError(ctx, "%+v", err)
		return
	}

	if total%packetsPerFile != 0 {
		log.V1.CtxInfo(ctx, "pk %d, packet total %d", id, total)
		return
	}

	go c.GenerateFile(ctx, unique, id)
}

func (c *compactor) GenerateFile(ctx context.Context, unique spec.RoomUnique, id int64) error {
	var retErr error
	span, ctx := bytedtracer.StartCustomSpan(ctx, trace.INTERNAL, trace.Compact,
		bytedtracer.AsAsyncChildSpan, bytedtracer.EnableEmitSpanMetrics)
	defer func() {
		span.SetTags(bytedtracer.NewTagKV(trace.IsError, retErr != nil))
		span.Finish()
		if retErr != nil {
			log.V1.CtxError(ctx, "GenerateFile fail,need check err:%+v params:appId:%s,roomId:%s,id:%d", retErr, unique.GetAppId(), unique.GetRoomId(), id)
		}
	}()

	packets, err := c.repo.GetPackets(ctx, unique, id)
	if err != nil {
		retErr = err
		return retErr
	}
	if len(packets) == 0 {
		retErr = errors.New("fail to get packets")
		return retErr
	}

	file := genFile(packets)
	fileName := genFileName(unique, packets[0].StartTime, id)

	retErr = c.upload(ctx, unique, fileName, file)
	if retErr != nil {
		return retErr
	}
	retErr = c.repo.SaveReplay(ctx, c.genReplay(unique, fileName, id, packets))
	return retErr
}

func (c *compactor) upload(ctx context.Context, unique spec.RoomUnique, name string, file []byte) error {
	vendor := object_storage.VendorVolcTos
	osClient, err := GetObjectStorage(ctx, vendor, c.accountID, c.bucket, c.region)
	if err != nil {
		log.V1.CtxError(ctx, "get object storage failed,os_type:%s,bucket:%s,error:%s", vendor, c.bucket, err)
		return err
	}
	err = osClient.Upload(ctx, name, bytes.NewReader(file), int64(len(file)), object_storage.ContentTypePB)
	if err != nil {
		log.V1.CtxError(ctx, "upload file failed,os_type:%s,bucket:%s,error:%s", vendor, c.bucket, err)
		return err
	}
	return nil
}

func (c *compactor) genReplay(unique spec.RoomUnique, fileName string, maxId int64, packets []*ddl.Packet) *ddl.Replay {
	actionCount := int32(0)
	for index := range packets {
		actionCount += packets[index].ActionCount
	}
	return &ddl.Replay{
		AppId:       unique.GetAppId(),
		RoomId:      unique.GetRoomId(),
		BoardId:     0,
		AccountId:   c.accountID,
		Bucket:      c.bucket,
		Region:      c.region,
		FileName:    fileName,
		StartTime:   packets[0].StartTime,
		EndTime:     packets[len(packets)-1].StartTime,
		MaxId:       maxId,
		ActionCount: actionCount,
	}
}

func genFile(packets []*ddl.Packet) []byte {
	totalLength := 0
	for _, pkt := range packets {
		totalLength += len(pkt.Data)
	}
	file := make([]byte, 0, totalLength)
	for _, pkt := range packets {
		file = append(file, pkt.Data...)
	}
	return file
}

func genFileName(unique spec.RoomUnique, startTime uint64, maxId int64) string {
	return fmt.Sprintf("%s_%s_%d_%d_%d",
		unique.GetAppId(), unique.GetRoomId(), 0, startTime, maxId)
}

func GetObjectStorage(ctx context.Context, vendor, accountID, bucket, region string) (object_storage.ObjectStorageClient, error) {
	cache := object_storage.GetCache()
	client := cache.GetClient(vendor, bucket)
	if client != nil {
		return client, nil
	}

	switch vendor {
	case object_storage.VendorVolcTos:
		endpoint := fmt.Sprintf(config.GetConfig().StorageConfig.VolcEndPoint, region)
		if env.IsBoe() {
			endpoint = fmt.Sprintf(config.GetConfig().StorageConfig.VolcEndPoint)
		}
		client, err := s3.New(ctx, s3.WithVolcConfig(&s3.VolcConfig{
			AccountId: accountID,
			Region:    region,
			EndPoint:  endpoint,
			Bucket:    bucket,
		}))
		if err != nil {
			return nil, errors.Wrap(err, "new object storage failed")
		}
		err = cache.SetClient(vendor, bucket, client)
		if err != nil {
			log.V1.CtxError(ctx, "set cache failed,error:%s", err)
		}
		return client, nil
	default:
		return nil, errors.New("object storage vendor not support")
	}

}
